import { readBody, defineEventHandler } from 'h3'
import redis from '../util/redis'
import { recordLog } from '../util/log'

// POST /api/chatstream2
// 说明：此接口将前端发送的 chatModelDto（或 messages 数组）转发到 dashscope 的
// compatible-mode chat completions 接口，并把上游的流式响应（SSE/text/event-stream）
// 透传回前端。建议在运行时通过环境变量 DASHSCOPE_KEY 提供 API Key。

export default defineEventHandler(async (event: any) => {
  const req = event.node.req
  const res = event.node.res

  // helper: get client IP (x-forwarded-for 优先)
  const getClientIp = (req: any) => {
    try {
      if (!req) return 'unknown'
      // Common headers that may contain the client IP
      const headersToCheck = ['x-forwarded-for', 'x-real-ip', 'x-client-ip', 'x-forwarded', 'forwarded-for', 'forwarded']
      for (const h of headersToCheck) {
        const val = req.headers && (req.headers[h] || req.headers[h.toUpperCase()])
        if (val) {
          const s = String(val).split(',')[0].trim()
          if (s) {
            // normalize IPv6-mapped IPv4
            return s.replace(/^::ffff:/, '')
          }
        }
      }

      // Some runtimes expose ip on connection/socket
      const addr = (req.socket && req.socket.remoteAddress) || (req.connection && (req.connection as any).remoteAddress) || (req.info && req.info.remoteAddress)
      if (addr) return String(addr).replace(/^::ffff:/, '')

      // Fallbacks used by some frameworks
      if ((req as any).ip) return (req as any).ip

      // give up
      try { console.warn && console.warn('[ip] failed to detect client ip, req.headers keys:', Object.keys(req.headers || {})) } catch (e) {}
      return 'unknown'
    } catch (e) {
      return 'unknown'
    }
  }
  // 必要的 SSE/流式响应头
  res.setHeader('Cache-Control', 'no-cache')
  res.setHeader('Content-Type', 'text/event-stream')
  res.setHeader('Connection', 'keep-alive')
  // 禁用某些代理的缓冲
  res.setHeader('X-Accel-Buffering', 'no')
  // 立即发送响应头，避免被某些环境缓冲
  try { res.flushHeaders && res.flushHeaders() } catch (e) { /* ignore */ }
  // 小函数：尝试调用可用的 flush 方法，强制把缓冲区发送到客户端
  const tryFlush = () => {
    try { if (typeof (res as any).flush === 'function') (res as any).flush() } catch (e) { /* ignore */ }
  }

  // 安全写入：在 writable 未结束时写入并捕获任何异常，避免在连接已关闭时抛出
  const safeWrite = (chunk: string) => {
    try {
      if (!res.writableEnded) res.write(chunk)
    } catch (e) {
      // ignore write errors (client disconnected etc.)
    }
  }

  const body = await readBody(event)

  // 兼容前端直接发送 messages 数组或发送包装对象 { messageDtos: [...] } 或 { messages: [...] }
  let payload: any = body
  if (Array.isArray(body)) {
    // 如果直接发送数组，按 OpenAI 风格包装为 messages
    payload = { messages: body }
  }
  // 如果前端发送了 messageDtos（旧后端可能需要），将其映射到 messages 以兼容 DashScope
  if (payload && payload.messageDtos && !payload.messages) {
    payload.messages = payload.messageDtos
  }

  // 强制使用指定模型，和原 C# 实现一致
  const MODEL = 'qwen-max-2025-01-25'
  try {
    if (payload && typeof payload === 'object') {
      payload.model = MODEL
      // 与原 C# 实现保持一致的默认参数
      // 温度（temperature）影响生成的随机性，0.3 保持较稳健的输出
      if (typeof payload.temperature === 'undefined') payload.temperature = 0.3
      // 启用上游流式返回
      if (typeof payload.stream === 'undefined') payload.stream = true
    }
  } catch (e) {
    // ignore
  }

  // 优先使用官方建议的环境变量名 DASHSCOPE_API_KEY，兼容旧名 DASHSCOPE_KEY
  const apiKey = process.env.DASHSCOPE_API_KEY || (body && body.apiKey) || ''
  if (!apiKey) {
    res.statusCode = 500
    safeWrite(`data: ${JSON.stringify({ error: 'Missing server API key. Set DASHSCOPE_KEY.', content: 'Missing server API key. Set DASHSCOPE_KEY.' })}\n\n`)
    await recordLog(event as any, { message: 'chatstream2: missing api key', userId: 0 })
    res.end()
    return
  }

  // 基于客户端 IP 限制：同一 IP 24 小时内只允许一次对话
  // 实现：在请求上游前尝试原子设置 Redis key (SET NX)，并使用短超时；
  // 如果 SET 返回 null 表示已存在（拒绝），如果超时或出错则记录并放行（避免阻塞流）。
  const clientIp = getClientIp(req)
  let bypassIpRateLimit = false
  const providedKey = (payload && (payload.randomKey || payload.key || payload.userKey)) || ''
  if (providedKey) {
    try {
      const k = String(providedKey)
      if (redis) {
        // 如果单 key 未命中，尝试在列表中查找（LPOS 更高效），LRANGE 回退
        if (!bypassIpRateLimit) {
          try {
            if (typeof (redis as any).lpos === 'function') {
              const pos = await (redis as any).lpos('random_keys', k)
              if (pos !== null && typeof pos !== 'undefined') bypassIpRateLimit = true
            } else if (typeof (redis as any).lrange === 'function') {
              const list = await (redis as any).lrange('random_keys', 0, -1)
              if (Array.isArray(list) && list.indexOf(k) !== -1) bypassIpRateLimit = true
            }
          } catch (e) {
            /* ignore list check error */
          }
        }

        if (bypassIpRateLimit) {
          try { console.info && console.info('[chatstream2] bypass ip rate limit for provided random_key') } catch (e) { /* ignore */ }
        }
      }
    } catch (e) {
      try { console.warn && console.warn('[redis] check random_key error', e) } catch (ee) { /* ignore */ }
    }
  }

  const redisKey = `chat:${clientIp}`
  try {
    if (!bypassIpRateLimit) {
      const redisSetNxWithTimeout = (key: string, value: string, exSeconds = 24 * 3600, timeout = 500) => {
        try {
          if (!redis || typeof (redis as any).set !== 'function') return Promise.resolve('NOREDIS')
          const p = (redis as any).set(key, value, 'EX', exSeconds, 'NX')
          const t = new Promise<'TIMEOUT'>((resolve) => setTimeout(() => resolve('TIMEOUT'), timeout))
          return Promise.race([p, t]) as Promise<any>
        } catch (e) {
          return Promise.resolve('ERR')
        }
      }

      // fire SET NX with a small value (we don't wait long)
      const setValue = JSON.stringify({ payload, savedAt: Date.now(), ip: clientIp })
      const setRes = await redisSetNxWithTimeout(redisKey, setValue, 24 * 3600, 500)
      if (setRes === null) {
        // key already exists
        res.statusCode = 429
        // 使用 Markdown 链接格式，前端可以按需以 Markdown 渲染为可点击链接
        const mdMsg = '[ 同一 IP 已在 24 小时内发送过对话，请稍后再试或者购买八八智能KEY（点击跳转八八小店）](/shop)。'
        safeWrite(`data: ${JSON.stringify({ error: mdMsg, content: mdMsg })}\n\n`)
        await recordLog(event as any, { message: `chatstream2: rate limited ip=${clientIp}`, userId: 0 })
        res.end()
        return
      } else if (setRes === 'TIMEOUT' || setRes === 'ERR' || setRes === 'NOREDIS') {
        try { console.warn && console.warn('[redis] set nx warning, continuing', setRes) } catch (ee) { /* ignore */ }
        // 继续，不阻塞请求
      }
    } else {
      try { console.info && console.info('[chatstream2] skipped IP rate limit because provided randomKey exists') } catch (e) { /* ignore */ }
    }
  } catch (e) {
    try { console.warn && console.warn('[redis] set nx error', e) } catch (ee) { /* ignore */ }
  }

  const upstreamUrl = 'https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions'
  const controller = new AbortController()

  // 当客户端断开时，取消上游请求
  const onClose = () => {
    try { controller.abort() } catch (e) { /* ignore */ }
  }
  req.on && req.on('close', onClose)

  try {
    const upstreamResp = await fetch(upstreamUrl, {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${apiKey}`,
        'Content-Type': 'application/json',
        // 明确期望 text/event-stream，以获得 SSE 格式的流式输出
        'Accept': 'text/event-stream, application/json'
      },
      body: JSON.stringify(payload),
      signal: controller.signal,
    })

    if (!upstreamResp.ok) {
      const txt = await upstreamResp.text().catch(() => '')
      // 以 SSE 格式返回错误信息
      safeWrite(`data: ${JSON.stringify({ error: `API请求失败: ${upstreamResp.status} ${upstreamResp.statusText}`, detail: txt, content: `API请求失败: ${upstreamResp.status} ${upstreamResp.statusText}` })}\n\n`)
      await recordLog(event as any, { message: `chatstream2: upstream error ${upstreamResp.status} ${upstreamResp.statusText}`, userId: 0 })
      res.end()
      return
    }

    // 如果上游没有 body（非流式），把完整响应透传回来
    if (!upstreamResp.body) {
      const txt = await upstreamResp.text().catch(() => '')
      safeWrite(`data: ${JSON.stringify({ content: txt })}\n\n`)
      await recordLog(event as any, { message: `chatstream2: upstream returned non-stream body`, userId: 0 })
      res.end()
      return
    }

  // upstreamResp.body 在不同环境中可能是多种可读流（Node Readable / WHATWG ReadableStream / event stream）
  const upstreamBody: any = upstreamResp.body

  // 不再发送 ': connected' 注释，避免出现在响应正文中
  try { tryFlush() } catch (e) { /* ignore */ }

  // 用于检测上游是否主动发送了 [DONE]
  let sawDone = false

    // 优先使用 async iterator（Node 18+ 等支持）
  if (upstreamBody && typeof upstreamBody[Symbol.asyncIterator] === 'function') {
      try {
        for await (const chunk of upstreamBody) {
          if (res.writableEnded) break
          let s: string
          // 处理各种 chunk 类型：Buffer / Uint8Array / ArrayBuffer / 其他
          if (typeof Buffer !== 'undefined' && Buffer.isBuffer(chunk)) {
            s = chunk.toString('utf8')
          } else if (chunk instanceof Uint8Array) {
            s = Buffer.from(chunk).toString('utf8')
          } else if (chunk instanceof ArrayBuffer) {
            s = Buffer.from(new Uint8Array(chunk)).toString('utf8')
          } else {
            s = String(chunk)
          }
          // 直接把上游的数据 chunk 透传给客户端（保留上游可能发送的 data: 前缀）
          safeWrite(s)
          if (s.indexOf('[DONE]') !== -1) sawDone = true
          tryFlush()
        }
      } catch (err) {
        // 上游或写入过程中出错，尝试告知客户端
        if (!res.writableEnded) {
            safeWrite(`data: ${JSON.stringify({ error: `流读取异常: ${String(err)}`, content: `流读取异常: ${String(err)}` })}\n\n`)
        }
      }
    } else if (upstreamBody && typeof upstreamBody.getReader === 'function') {
      // WHATWG ReadableStream
      try {
        const reader = upstreamBody.getReader()
        while (true) {
          const { done, value } = await reader.read()
          if (done) break
          if (res.writableEnded) break
          let s: string
          if (typeof Buffer !== 'undefined' && Buffer.isBuffer(value)) {
            s = value.toString('utf8')
          } else if (value instanceof Uint8Array) {
            s = Buffer.from(value).toString('utf8')
          } else if (value instanceof ArrayBuffer) {
            s = Buffer.from(new Uint8Array(value)).toString('utf8')
          } else {
            s = String(value)
          }
          safeWrite(s)
          if (s.indexOf('[DONE]') !== -1) sawDone = true
          tryFlush()
        }
      } catch (err) {
        if (!res.writableEnded) {
          safeWrite(`data: ${JSON.stringify({ error: `流读取异常: ${String(err)}`, content: `流读取异常: ${String(err)}` })}\n\n`)
        }
      }

    } else if (upstreamBody && upstreamBody.on) {
      // 兼容使用事件的 stream
      await new Promise<void>((resolve, reject) => {
        upstreamBody.on('data', (chunk: any) => {
          if (res.writableEnded) return
          let s: string
          if (typeof Buffer !== 'undefined' && Buffer.isBuffer(chunk)) {
            s = chunk.toString('utf8')
          } else if (chunk instanceof Uint8Array) {
            s = Buffer.from(chunk).toString('utf8')
          } else if (chunk instanceof ArrayBuffer) {
            s = Buffer.from(new Uint8Array(chunk)).toString('utf8')
          } else {
            s = String(chunk)
          }
          safeWrite(s)
          if (s.indexOf('[DONE]') !== -1) sawDone = true
          tryFlush()
        })
        upstreamBody.on('end', () => resolve())
        upstreamBody.on('error', (err: any) => reject(err))
        res.on('close', () => {
          // 客户端断开
          try { controller.abort() } catch (e) {}
          resolve()
        })
      })
    }

    // 如果上游未发送 [DONE]，主动发送一个以提示客户端流已完成
    try {
      if (!sawDone && !res.writableEnded) {
        safeWrite('data: [DONE]\n\n')
      }
    } catch (e) { /* ignore */ }
    // NOTE: 我们已在请求开始前做过 SET NX，终结时不再等待或写入 Redis，以避免阻塞流。

    // 结束响应
    if (!res.writableEnded) res.end()
    try { await recordLog(event as any, { message: `chatstream2: completed ip=${clientIp}`, userId: 0 }) } catch (e) { /* ignore */ }
  } catch (err: any) {
    if (!res.writableEnded) {
      safeWrite(`data: ${JSON.stringify({ error: `处理请求时发生错误: ${String(err?.message || err)}`, content: `处理请求时发生错误: ${String(err?.message || err)}` })}\n\n`)
      await recordLog(event as any, { message: `chatstream2: handler error ${String(err?.message || err)}`, userId: 0 })
      res.end()
    }
  } finally {
    // 清理监听器
    try { req.off && req.off('close', onClose) } catch (e) { /* ignore */ }
  }
})
